package com.example.dayu521.sg_oss.controller;

import cn.hutool.poi.excel.ExcelUtil;
import com.alibaba.cloud.stream.binder.rocketmq.constant.RocketMQConst;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.write.metadata.WriteSheet;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.example.dayu521.sg_oss.domain.Trade;
import com.example.dayu521.sg_oss.domain.donain_mid.CartParam;
import com.example.dayu521.sg_oss.domain.donain_mid.PageBase;
import com.example.dayu521.sg_oss.domain.donain_mid.TradeParam;
import com.example.dayu521.sg_oss.service.Kl;
import com.example.dayu521.sg_oss.service.PSettleService;
import com.example.dayu521.sg_oss.service.TradeService;
import com.example.dayu521.sg_oss.service.feign.CartService;
import com.example.dayu521.sg_oss.service.feign.PayServiceRpc;
import com.example.dayu521.sg_oss.service.stream.EmitterOPS;
import jakarta.servlet.http.HttpServletResponse;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.example.dayu521.OrRes;
import org.example.dayu521.OrderState;
import org.example.dayu521.exception.NoSuchOss;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.core.io.InputStreamResource;
import org.springframework.http.ContentDisposition;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;

import java.io.*;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;

@RestController
@RequestMapping("/trade")
@Slf4j
public class ProductTrade {

    @Autowired
    private StreamBridge streamBridge;

    @Autowired
    private CartService cartService;

    //考虑到tcc的复杂性,我们也可以使用事务消息,提前埋点,对于不能加入at事务的操作,待消息队列回查会最终把at事务外的操作恢复
    @PostMapping("/settle")
    public OrRes<?> sattle(@RequestHeader("login_id") String login_id) throws Exception {

        var ox = new Kl<OrRes<?>, TradeParam>() {

            public final Px<TradeParam, ArrayBlockingQueue<OrRes<?>>, Exception> f = new Px();

            @Override
            public boolean set_transaction_post_trigger(TradeParam s) {
                var builder = MessageBuilder.withPayload(s.getSettle_id());
                builder.setHeader("test", 0)
                        .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
                //监听器的arg参数将收到这个参数
                f.setP(s);
                f.setR(new ArrayBlockingQueue<OrRes<?>>(1));
                f.setE(null);
                builder.setHeader(RocketMQConst.USER_TRANSACTIONAL_ARGS, f);
                var msg = builder.build();
                streamBridge.send("transaction_trigger-out-0", msg);
                System.out.println("send Msg:" + msg.toString());
                return true;
            }

            @Override
            public OrRes<?> await_transaction() throws Exception {
                //单线程,没有超时异常
                var res = f.getR().poll(30, TimeUnit.SECONDS);
                if (f.getE() != null) {
                    throw f.getE();
                }
                if (res == null) {
                    throw new TimeoutException("服务调用超时");
                }
                return res;
            }
        };

        ////todo 购物车直接防重了,这里需要的是接口保护,更进一步的是针对用户这种特定情况下的业务保护,一般不使用sentinel,他是更底层的接口保护

        CartParam cartParam = new CartParam();
        cartParam.setLogin_id(login_id);
        if (!cartService.cart_state(cartParam).is_ok()) {
            return OrRes.error("购物车没有商品");
        }

        var tradeParam = new TradeParam();
        tradeParam.setLogin_id(login_id);
        tradeParam.setSettle_id(UUID.randomUUID().toString());

        ox.set_transaction_post_trigger(tradeParam);

        return ox.await_transaction();
    }

    @Autowired
    private PSettleService pSettleService;

    @PostMapping("/settle2")
    public OrRes<?> sattle2(@RequestHeader("login_id") String login_id) throws Exception {
        CartParam cartParam = new CartParam();
        cartParam.setLogin_id(login_id);
        if (!cartService.cart_state(cartParam).is_ok()) {
            return OrRes.error("购物车没有商品");
        }
        var tradeParam = new TradeParam();
        tradeParam.setLogin_id(login_id);
        tradeParam.setSettle_id(UUID.randomUUID().toString());

        pSettleService.order_trade(tradeParam);

        return OrRes.ok();
    }

    @Autowired
    private TradeService tradeService;

    //可能返回空
    @PostMapping("/get_order")
    public OrRes<Trade> get_order(@RequestBody Long order_id) {
        var one = tradeService.lambdaQuery().
                eq(Trade::getBusiness_id, order_id).
                eq(Trade::getState, 1).
                one();
        if (one == null) {
            return OrRes.error("没有此订单").x();
        }
        return OrRes.ok(one);
    }

    @Value("${sg-oss.order.timeout}")
    private Long timeout;

    @PostMapping("/get_order2")
    public OrRes<Trade> get_order2(@RequestBody Long order_id) {
        var one = get_order(order_id);
        Trade res = one.res();
        if (res != null) {
            res.setTimeout_date(res.getDate().plusSeconds(timeout));
        }
        return one;
    }


    @PostMapping("/order/page")
    public OrRes<Page<Trade>> order_list(@RequestHeader("login_id") String login_id, @RequestBody PageBase page) {
        var one = tradeService.lambdaQuery().
                eq(Trade::getUser_id, login_id).
                eq(Trade::getState, 1).
                orderByDesc(Trade::getDate).
                page(
                        Page.of(
                                Optional.ofNullable(page.getCurrent_page()).orElse(1l),
                                Optional.ofNullable(page.getPage_size()).orElse(7l)
                                , false)
                );
        return OrRes.ok(one);
    }

    @PostMapping("/order/page/download")
    public ResponseEntity<ResponseBodyEmitter> handle(@RequestHeader("login_id") String login_id, @RequestBody PageBase page/*, HttpServletResponse hsr*/) {
        var emitter = new ResponseBodyEmitter();
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            executor.submit(() -> {
                try {
                    var next_index = 1l;
                    var size = 5l;
                    page.setPage_size(size);

                    try (var writer = ExcelUtil.getWriter(true)) {
                        while (true) {
                            try {
                                var list = order_list(login_id, page).res();
                                if (list.getRecords().size() == 0) {
                                    try (var some_output = new ByteArrayOutputStream()) {
                                        writer.flush(some_output, false);
                                        emitter.send(some_output.toByteArray());
                                    }
                                    emitter.complete();
                                    break;
                                }
                                writer.write(list.getRecords());
//                                Thread.sleep(1000);

                                //BUG 多次flush,则会显著增大文件大小
//                                try (var some_output = new ByteArrayOutputStream()) {
//                                    writer.flush(some_output, false);
//                                    emitter.send(some_output.toByteArray());
//                                }

                                //检测客户端是否关闭
                                emitter.send("".getBytes());
                            } catch (IOException e) {
                                //用户中断连接
                                throw e;
                            }
                            next_index++;
                            page.setCurrent_page(next_index);
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
            });
        }

        return ResponseEntity.ok().headers(a -> {
            a.setContentType(MediaType.APPLICATION_OCTET_STREAM);
            a.setContentDisposition(ContentDisposition.attachment().build());
        }).body(emitter);

    }

    @PostMapping("/order/page/download2")
    public ResponseEntity<ResponseBodyEmitter> handle2(@RequestHeader("login_id") String login_id, @RequestBody PageBase page/*, HttpServletResponse hsr*/) {
        var op = new EmitterOPS();
        var emitter = op.getRbe();
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            executor.submit(() -> {
                try {
                    var next_index = 1l;
                    var size = 5l;
                    page.setPage_size(size);

                    WriteSheet writeSheet = EasyExcel.writerSheet().build();
                    try (var writer = EasyExcel.write(op,Trade.class).autoCloseStream(true).build()) {
//                    try (op;var writer = ExcelUtil.getWriter(true)) {
                        while (true) {

                            var list = order_list(login_id, page).res();
                            if (list.getRecords().size() == 0) {
//                                writer.flush(op);
                                break;
                            }

//                            writer.write(list.getRecords());
                            writer.write(list.getRecords(),writeSheet);
                            next_index++;
                            page.setCurrent_page(next_index);
                        }
                    }
                } catch (Exception e) {
                    log.error(e.getMessage());
                }
            });
        }

        return ResponseEntity.ok().headers(a -> {
            a.setContentType(MediaType.APPLICATION_OCTET_STREAM);
            a.setContentDisposition(ContentDisposition.attachment().build());
        }).body(emitter);

    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static final class FinishTrade {
        private String order_no;
        private String order_state;
    }

    //前提是订单处于创建中.然后由支付服务完成后调用,竞争与冲突被支付服务解决掉了
    @PostMapping("/order/finish_pay")
    public OrRes<?> finish_pay(@RequestBody FinishTrade finishTrade) {
        //需要判断
        if (OrderState.valueOf(finishTrade.order_state) == OrderState.Payed) {
            var update = tradeService.lambdaUpdate().eq(Trade::getBusiness_id, finishTrade.order_no).
                    and(self -> self.eq(Trade::getBs_state, 0).or().eq(Trade::getBs_state, 1)).
                    set(Trade::getBs_state, order_state_map.get(finishTrade.order_state)).update();
            log.info(STR."订单收到完成通知 \{finishTrade.order_no},调用结果 \{update}");
        }
        return OrRes.ok(finishTrade.order_state);
    }

    @Autowired
    private PayServiceRpc payServiceRpc;

    @PostMapping("/order/close")
    public OrRes<?> close_order(@RequestBody Long order_id) {
        return close_order(order_id, OrderState.Canceled, false);
    }

    //在主动关单时,我们仍旧不能决定三方支付服务的状态.
    // 如果设置force,则认为订单在此时的状态已经确定.我们也会发送一个退款消息来尝试补偿已经支付成功的通知
    public OrRes<String> close_order(Long order_id, OrderState r, boolean force) {

        //提前进行查看是否需要关闭
        Trade one = tradeService.lambdaQuery().eq(Trade::getBusiness_id, order_id).
                eq(Trade::getState, 1).one();
        if (one == null) {
            return OrRes.ok(OrderState.NotFound.getValue());
        } else if (one.getBs_state() == 1) {
            return OrRes.ok(OrderState.Payed.getValue());
        } else if (one.getBs_state() == 2) {
            return OrRes.ok(OrderState.Canceled.getValue());
        } else if (one.getBs_state() == 3) {
            return OrRes.ok(OrderState.Timeout.getValue());
        } else if (one.getBs_state() == 0 || (one.getBs_state() == 4 && force)) {//状态为0 可以进行关单

            //todo 竞争在远程服务解决,但这里仍需判断
            // 远程关单,竞争都在远程服务进行决定,
            var close_res = payServiceRpc.close_order(new PayServiceRpc.CloseOrderParam(order_id.toString(), force, r.getValue()));
            OrderState orderState = OrderState.valueOf(close_res.res().getOrder_state());
            if (orderState == OrderState.Payed || orderState == OrderState.Canceled) {
                //更新失败说明没有此订单
                @NonNull var val = order_state_map.get(close_res.res().getOrder_state());
                if (orderState == OrderState.Canceled) {
                    val = order_state_map.get(close_res.res().getClose_reason());
                }
                var x = val;
                Objects.nonNull(x);
                //如果存在下面条件的数据,就不会失败.
                //如果失败,不用管,说明此时是过期调用
                tradeService.lambdaUpdate().eq(Trade::getBusiness_id, order_id).
                        eq(Trade::getState, 1).
                        and(self -> self.eq(Trade::getBs_state, 0).or().eq(Trade::getBs_state, 4).or().eq(Trade::getBs_state, x))
                        .set(Trade::getBs_state, val).update();
                //todo 返回查询状态
            }
            return OrRes.ok(close_res.res().getOrder_state());

        } else if (one.getBs_state() == 4) {
            return OrRes.ok(OrderState.Cancel.getValue());
        } else {
            throw new RuntimeException("未实现");
        }
    }

    //map不可修改
    private static final Map<String, Integer> order_state_map = Map.of(
            OrderState.Created.getValue(), 0,
            OrderState.Payed.getValue(), 1,
            OrderState.Canceled.getValue(), 2,
            OrderState.Timeout.getValue(), 3,
            OrderState.Cancel.getValue(), 4,
            OrderState.NotFound.getValue(), 0 //如果没有发现,还是作为创建状态
    );

    //决定订单创建后的下一个状态.
    @PostMapping("/order/decide_state")
    @NonNull
    public OrRes<String> decide_state_after_created(@RequestBody PayServiceRpc.TradeStateParam tradeStateParam) {
        Trade one = tradeService.lambdaQuery().eq(Trade::getBusiness_id, Long.parseLong(tradeStateParam.getOrder_no())).one();
        if (one == null) {
            return OrRes.error_res(OrderState.NotFound.getValue());
        } else if (one.getBs_state() == 0 || one.getBs_state() == 4) {//此时状态未确定
            //这里返回的状态只能是 cancel payed canceled,
            //因为其他的状态前提是 当前状态不是created(0)或者cancel(4),所以在当前状态的前提下,他们会插入失败

            //同时,如果返回的状态是cancel,则一定还会返回payed或canceled,
            //且payed和canceled是不可能同时返回,如果返回他们其中一个,则即使返回cancel,因为当前状态改为payed和canceled,所以cancel会插入失败

            OrRes<String> orRes = payServiceRpc.poll_trade_state(tradeStateParam);

            // 竞争在远程服务解决,但这里仍需判断
            @NonNull var val = order_state_map.get(orRes.res());
            tradeService.lambdaUpdate().eq(Trade::getBusiness_id, Long.parseLong(tradeStateParam.getOrder_no())).
                    and(self -> {
                        self.eq(Trade::getBs_state, 4).or().eq(Trade::getBs_state, 0);
                    }).
                    set(Trade::getBs_state, val).update();

            return orRes;
        } else if (one.getBs_state() == 1) {
            return OrRes.ok(OrderState.Payed.getValue());
        } else if (one.getBs_state() == 2) {
            return OrRes.ok(OrderState.Canceled.getValue());
        } else if (one.getBs_state() == 3) {
            return OrRes.ok(OrderState.Timeout.getValue());
        } else if (one.getBs_state() == 5) {
            return OrRes.ok(OrderState.NotFound.getValue());
        } else {
            throw new RuntimeException("未实现");
        }
    }

}
